package com.jzsec.tzdslog;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class TzdsLoginLogProcessJob {

    private static ObjectMapper mapper = new ObjectMapper();

    public static void main(String[] args) throws Exception {

        final ParameterTool argsParamTool = ParameterTool.fromArgs(args);
        String logdate = argsParamTool.get("logdate", "2020-01-16"); // --logdate 2019-10-15
        String filePrefix = "";  // --isRzrq 1
        final Integer isRzrq = argsParamTool.getInt("isRzrq", 0);
        if (1 == isRzrq) {
            filePrefix = "lr";
        }
        final String propertiesFile = argsParamTool.get("conf", "conf.properties");
        final ParameterTool propParamTool = ParameterTool.fromPropertiesFile(propertiesFile);

        // 创建上下文
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(propParamTool);

        DataStreamSource<CCustLoginHisEntity> dataStreamSource = env.addSource(
                new CCustLoginHisSource(
                        propParamTool.get("mysql.jdbc.url"),
                        propParamTool.get("mysql.jdbc.username"),
                        propParamTool.get("mysql.jdbc.password"),
                        LocalDate.parse(logdate))
        );
        SingleOutputStreamOperator<String> logStream = dataStreamSource.map(entity -> {
            try {
                LogModel log = new LogModel();
                log.setAppType("tzdsAccess");
                log.setCid("0x1C030000");
                log.setMid("0x1C030001");
                log.setName("tzds");
                log.setIp(entity.getUser_ip());
                log.setIndicator(entity2Map(entity));
                log.setTime_stamp(Long.toString(entity.getRequest_time().getTime()));
                return JSONObject.toJSONString(log);
            } catch (Exception e) {
                return "";
            }
        });
        SingleOutputStreamOperator<String> unifyLogStream = dataStreamSource.map(entity -> {
            try {
                Map unifyLog = new HashMap();
                initLoginLog(unifyLog);
                procLoginLog(unifyLog, entity2Map(entity), Long.toString(entity.getRequest_time().getTime()));
                return JSONObject.toJSONString(unifyLog);
            } catch (Exception e) {
                return "";
            }
        });

        logStream.addSink(new SinkToTextFile(
                propParamTool.get("log.filepath") + "/" + logdate,
                "tzdslogin_" + filePrefix + "uf_" + logdate.replace("-", "") + ".log"
        ));
        unifyLogStream.addSink(new SinkToTextFile(
                propParamTool.get("unifyLog.filepath") + "/" + logdate,
                "tzdslogin_" + filePrefix + "his_" + logdate.replace("-", "") + ".log"
        ));
        // 写入kafka
        // Properties kafkaProps = new Properties();
        // kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, propParamTool.get("kafka.bootstrap.servers"));
        // kafkaProps.put("client.id", propParamTool.get("kafka.client.id"));
        // kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
        // kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
        // String topic = propParamTool.get("kafka.topic.loginbycustid");
        // unifyLogStream.addSink(new SinkToKafka011(kafkaProps, topic)).name("sink-to-kafka").setParallelism(1);

        env.execute();
    }

    private static Map entity2Map(CCustLoginHisEntity entity) {
        Map map = new HashMap();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        map.put("accessTime", sdf.format(entity.getRequest_time()));
        map.put("httpInfo", "POST /cuser/loginbycustid HTTP/1.1");
        map.put("httpRetCode", "200");
        map.put("contentLength", 78);
        map.put("agentInfo", entity.getUser_agent());
        map.put("clientIp", getClientIp("", entity.getUser_ip()));
        map.put("upstream_rsp_time", entity.getElapsed_time());
        map.put("request_time", entity.getElapsed_time());

        map.put("custid", entity.getCust_id());
        map.put("sessionId", entity.getSession_id());
        map.put("deviceCode", entity.getDevice_code());
        map.put("deviceType", entity.getDevice_type());
        map.put("appVer", entity.getApp_version());
        map.put("appType", entity.getApp_type());
        map.put("envType", entity.getEnv_type());
        map.put("trade_url", entity.getTrade_url());

        String opStation = entity.getOp_station();
        if (StringUtils.isNotEmpty(opStation)) {
            String[] arr = opStation.split("\\|");
            map.put("opstat_clientip", OpstationUtils.formatIp(arr[1]));
            map.put("mac", OpstationUtils.formatMac(arr[2]));
            map.put("mobilephone", OpstationUtils.formatMobilephone(arr[5]));
            map.put("imei", OpstationUtils.formatImei(arr[10]));
            map.put("udid", OpstationUtils.formatUdid(arr[12]));
        }
        map.put("responseMsg", entity.getResponse_msg());
        return map;
    }

    private static String procLog(String line) throws ParseException {
        String clientIp = "";
        LogModel log = new LogModel();
        Map map = new HashMap();
        Map<String, String> unifyLog = new HashMap();
        initLoginLog(unifyLog);

        log.setAppType("tzdsAccess");
        log.setCid("0x1C030000");
        log.setMid("0x1C030001");
        log.setName("tzds");
        log.setIndicator(map);

        Pattern p = Pattern.compile("(\\S+)\\s");
        Matcher m = p.matcher(line);
        if (m.find()) {
            clientIp = m.group(0);
            log.setIp("");
        }

        try {
            String timeStr = line.substring(line.indexOf("[") + 1, line.lastIndexOf("]")).replace("]", "");
            SimpleDateFormat sdf = new SimpleDateFormat("d/MMM/yyyy:K:m:s Z", Locale.ENGLISH);
            SimpleDateFormat sdfnew = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            Date d2 = null;
            try {
                d2 = sdf.parse(timeStr.trim());
            } catch (ParseException e) {
                return "";
            }
            timeStr = sdfnew.format(d2);
            log.setTime_stamp(Long.toString(d2.getTime()));
            map.put("accessTime", timeStr);


            String method = line.substring(line.indexOf("]") + 1, line.length());
            Pattern pm = Pattern.compile("^\\s\"(.*?)\"\\s(\\d+)\\s(\\d+)\\s\".*?\"\\s\"(.*?)\"\\s\"(.*?)\"\\s\"(.*?)\"\\s(.*?)\\s(.*?)\\s(.*?)\\s(.*?)$");
            Matcher mm = pm.matcher(method);
            if (mm.find()) {
                map.put("httpInfo", mm.group(1));
                map.put("httpRetCode", mm.group(2));
                map.put("contentLength", mm.group(3));
                map.put("agentInfo", mm.group(5));
                map.put("clientIp", getClientIp(mm.group(6), clientIp));
                map.put("upstream_rsp_time", mm.group(8));
                map.put("request_time", mm.group(10));
                procRequestParams(map, mm.group(4));
                procLoginLog(unifyLog, map, log.getTime_stamp());

                //				for (Map.Entry entry : unifyLog.entrySet()) {
                //					System.out.print(entry.getKey() + " => " + entry.getValue() + ", ");
                //				}
                //				System.out.println();
                return mapper.writeValueAsString(unifyLog);
            }
        } catch (Exception e) {
            System.out.println(e);
            return "";
        }
        return "";
    }

    private static String getClientIp(String proxyIP, String clientIP) {
        if (proxyIP.equals("-")) {
            return clientIP.replace(" ", "");
        }
        if (clientIP.contains("10.10") || clientIP.contains("10.13") || clientIP.contains("10.198")) {
            return proxyIP;
        }
        return clientIP.replace(" ", "");
    }

    private static void procRequestParams(Map map, String params) {
        if (params.equals("-")) {
            return;
        }
        String param = params.replace("\\x22", "\"");
        JSONObject jsonObj = JSON.parseObject(param.trim());
        for (Map.Entry<String, Object> entry : jsonObj.entrySet()) {
            if (((String) entry.getKey()).equals("op_station")) {
                String[] station = ((String) entry.getValue()).split("\\|");
                for (int j = 0; j < station.length; j++) {
                    String[] stationItem = station[j].contains("mac") ? station[j].split(":", 2) : station[j].split(":");
                    if (stationItem.length == 2)
                        if (stationItem[0].equals("clientip")) {
                            map.put("opstat_clientip", stationItem[1]);
                        } else {
                            map.put(stationItem[0], stationItem[1]);
                        }
                }
                continue;
            }
            if (((String) entry.getKey()).equals("password"))
                continue;
            if (((String) entry.getKey()).equals("login_info")) {
                map.put("input_content", ((JSONObject) entry.getValue()).getString("input_content"));
                continue;
            }
            map.put(entry.getKey(), entry.getValue());
        }
    }

    private static void initLoginLog(Map map) {
        map.put("brokerName", "九州证券");
        map.put("branchNo", "");
        map.put("branchName", "");
        map.put("fundId", "");
        map.put("secName", "");
        map.put("operDate", "");
        map.put("operTime", "");
        map.put("operDesc", "");
        map.put("entrustWay", "");
        map.put("entrustWayName", "手机炒股");
        map.put("appTypeName", "");
        map.put("appType", "");
        map.put("appID", "TZDS");
        map.put("appIDName", "投资大师");
        map.put("appVer", "");
        map.put("computerName", "");
        map.put("clientIP", "");
        map.put("mac", "");
        map.put("cpuNO", "");
        map.put("hddCode", "");
        map.put("hddLvInfo", "");
        map.put("imei", "");
        map.put("uuid", "");
        map.put("guid", "");
        map.put("osVer", "");
        map.put("mobilePhone", "");
        map.put("phone", "");
        map.put("bpRetCode", "");
        map.put("status", Integer.valueOf(0));
        map.put("timeStamp", "");
    }

    private static void procLoginLog(Map unify, Map map, String time_stamp) {
        unify.put("timeStamp", time_stamp);
        unify.put("branchNo", (map.get("branch_no") != null) ? map.get("branch_no") : "");
        unify.put("fundId", (map.get("custid") != null) ? map.get("custid") : map.get("input_content"));
        unify.put("operDate", (map.get("accessTime") != null) ? ((String) map.get("accessTime")).split(" ")[0] : "");
        unify.put("operTime", (map.get("accessTime") != null) ? ((String) map.get("accessTime")).split(" ")[1] : "");
        unify.put("entrustWay", (map.get("entrust_way") != null) ? map.get("entrust_way") : "");
        unify.put("appType", (map.get("deviceType") != null && map.get("deviceType").equals("IOS")) ? "MI" : "MA");
        if (unify.get("appType").equals("MI")) {
            unify.put("appTypeName", "移动IOS终端");
        } else {
            unify.put("appTypeName", "移动Android终端");
        }
        unify.put("appVer", (map.get("ver") != null) ? map.get("ver") : map.get("appVer"));
        unify.put("computerName", (map.get("computername") != null) ? map.get("computername") : "");
        unify.put("clientIP", (map.get("clientIp") != null) ? map.get("clientIp") : "");
        unify.put("mac", (map.get("mac") != null) ? map.get("mac") : "");
        unify.put("imei", (map.get("imei") != null) ? map.get("imei") : "");
        unify.put("uuid", (map.get("udid") != null) ? map.get("udid") : map.get("deviceCode"));
        unify.put("osVer", (map.get("agentInfo") != null) ? map.get("agentInfo") : "");
        unify.put("mobilePhone", (map.get("mobilephone") != null) ? map.get("mobilephone") : "");
    }

}
